package demo_day3;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;

public class TestMyPartitioner {
    public static void main(String[] args) {
        HashMap<String,Object> confMap = new HashMap<>();
        confMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop11:9092,hadoop12:9092,hadoop13:9092");
        confMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        confMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // 如果你写了一个自定义分区，但是没有做任何配置操作，那么你的发送者还会根据系统进行分区的发送
        confMap.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"demo_day3.Mypartitioner");

        KafkaProducer<String,String> producer = new KafkaProducer<>(confMap);

        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("topicB","hello kafka" + i));
        }
        producer.send(new ProducerRecord<>("topicB","嘿嘿嘿"));

        // 4、关闭资源
        producer.close();
    }
}
